from pyspark.sql import SparkSession, Window
import os
import pyspark.sql.functions as F

from cn.itcast.tag.base.BaseModel import BaseModel
from cn.itcast.tag.bean.ESMeta import ruleToESMeta

"""
-------------------------------------------------
   Description :	TODO：
   SourceFile  :	PayWayModel
   Author      :	itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'

#todo 支付方式
class PayWayModel(BaseModel):
    def compute(self, es_df, five_df):
        """
        标签计算逻辑：根据用户支付方式统计并匹配五级标签

        Args:
            es_df: 业务数据，包含支付记录信息
            five_df: 五级标签规则数据

        Returns:
            DataFrame: 计算结果DataFrame，包含userId和tagsId字段
        """
        # 对用户按memberid和paymentcode进行分组，并统计每个支付方式的使用次数
        es_df = es_df.groupby("memberid", "paymentcode").agg(
            F.count("paymentcode").alias("paymentcode_cnt")
        )

        # 使用row_number开窗函数为每个用户的支付方式进行排名（按支付次数降序）
        es_df = es_df.select(
            "memberid",
            "paymentcode",
            "paymentcode_cnt",
            F.row_number()
            .over(Window.partitionBy("memberid").orderBy(F.desc("paymentcode_cnt")))
            .alias("rn"),
        )

        # 筛选出每个用户排名第一的支付方式（即最常用的支付方式）
        es_df = es_df.where("rn = 1").select("memberid", "paymentcode")

        # 展示中间结果（调试用）
        es_df.show()

        # 将处理后的数据与五级标签规则进行左连接，匹配对应的标签ID
        new_df = es_df.join(
            five_df, es_df["paymentcode"] == five_df["rule"], "left"
        ).select(
            es_df["memberid"].alias("userId"), five_df["id"].alias("tagsId")
        )

        # 展示五级标签数据和最终计算结果（调试用）
        five_df.show()
        new_df.show()

        # 返回计算结果
        return new_df


if __name__ == '__main__':
    model = PayWayModel(30)
    model.execute()
